package cn.getech.data.development.service.impl;

import cn.getech.data.development.config.properties.BdpJobConfig;
import cn.getech.data.development.config.properties.DataDevelopmentConfig;
import cn.getech.data.development.constant.RealTaskAlarmModeEnum;
import cn.getech.data.development.dto.IdAndNameDto;
import cn.getech.data.development.entity.*;
import cn.getech.data.development.mapper.BdpRealTaskAlarmMapper;
import cn.getech.data.development.mapper.RealTimeTaskMapper;
import cn.getech.data.development.model.dto.RealTaskAlarmInfoDto;
import cn.getech.data.development.model.dto.RealTaskAlarmSaveDto;
import cn.getech.data.development.model.vo.IdAndNameVO;
import cn.getech.data.development.model.vo.RealTaskAlarmSearchVO;
import cn.getech.data.development.model.vo.RealTaskAlarmVO;
import cn.getech.data.development.model.vo.RealTimeTaskVO;
import cn.getech.data.development.service.*;
import cn.getech.data.development.utils.SendEMailUtil;
import cn.getech.data.development.utils.hadoop.HadoopLogUtils;
import cn.getech.data.development.utils.shell.ExecuteShellUtil;
import cn.getech.data.intelligence.common.exception.RRException;
import cn.getech.data.intelligence.common.utils.*;
import cn.getech.system.center.entity.SysUserEntity;
import cn.getech.system.center.model.res.UserListRes;
import cn.getech.system.center.service.SysUserService;
import cn.getech.system.center.utils.ShiroUtils;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.map.MapUtil;
import cn.hutool.http.HttpUtil;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.google.common.base.Joiner;
import com.jcraft.jsch.JSchException;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;
import java.time.LocalDateTime;
import java.time.LocalTime;
import java.time.ZonedDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;

/**
 * <p>
 * 实时任务告警表 服务实现类
 * </p>
 *
 * @author hujz
 * @since 2021-04-08
 */
@Slf4j
@Service
public class BdpRealTaskAlarmServiceImpl extends ServiceImpl<BdpRealTaskAlarmMapper, BdpRealTaskAlarm> implements IBdpRealTaskAlarmService {
    @Autowired
    private BdpRealTaskAlarmMapper bdpRealTaskAlarmMapper;
    @Resource
    private RealTimeTaskMapper realTimeTaskMapper;
    @Autowired
    private IBdpRealTaskAlarmReceivedService iBdpRealTaskAlarmReceivedService;
    @Autowired
    private IBdpRealTaskAlarmObjectService iBdpRealTaskAlarmObjectService;
    @Autowired
    private IBdpRealTaskAlarmSendLogService iBdpRealTaskAlarmSendLogService;
    @Autowired
    private ProcInfoService procInfoService;
    @Autowired
    private SendEMailUtil sendEMailUtil;
    @Autowired
    private SysUserService sysUserService;
    @Autowired
    private DataDevelopmentConfig developmentConfig;
    @Autowired
    private BdpJobConfig bdpJobConfig;
    @Autowired
    private RealTimeTaskMenuService realTimeTaskMenuService;

    @Override
    public PageUtils alarmPage(RealTaskAlarmSearchVO realTaskAlarmSearchVO) {
        //全部的触发条件  0:失败
//        if (Objects.equals(0, realTaskAlarmSearchVO.getTriggerCondition())) {
//            realTaskAlarmSearchVO.setTriggerCondition(null);
//        }
        realTaskAlarmSearchVO.setTenantId(ShiroUtils.getTenantId());
        IPage<BdpRealTaskAlarm> page = bdpRealTaskAlarmMapper
                .alarmPage((Page<BdpRealTaskAlarm>) new QueryBean<BdpRealTaskAlarm, RealTaskAlarmSearchVO>().getPage(realTaskAlarmSearchVO)
                        , realTaskAlarmSearchVO);
        if (CollectionUtil.isNotEmpty(page.getRecords())) {
            List<Integer> alarmIds = page.getRecords().stream().map(BdpRealTaskAlarm::getId).distinct().collect(Collectors.toList());
            List<BdpRealTaskAlarmReceived> receiveds = iBdpRealTaskAlarmReceivedService.listAndNameByIds(alarmIds);
            Map<Integer, List<String>> receivedMap = new HashMap<>();
            List<String> receivedNameList;
            if (CollectionUtil.isNotEmpty(receiveds)) {
                for (BdpRealTaskAlarmReceived received : receiveds) {
                    receivedNameList = receivedMap.get(received.getAlarmId());
                    receivedNameList = receivedNameList == null ? new ArrayList<>() : receivedNameList;
                    receivedNameList.add(received.getReceivedName());
                    receivedMap.put(received.getAlarmId(), receivedNameList);
                }
            }
            List<BdpRealTaskAlarmObject> alarmObjects = iBdpRealTaskAlarmObjectService.listAndNameByIds(alarmIds);
            Map<Integer, List<String>> alarmObjectMap = new HashMap<>();
            List<String> objectNameList;
            if (CollectionUtil.isNotEmpty(alarmObjects)) {
                for (BdpRealTaskAlarmObject object : alarmObjects) {
                    objectNameList = alarmObjectMap.get(object.getAlarmId());
                    objectNameList = objectNameList == null ? new ArrayList<>() : objectNameList;
                    objectNameList.add(object.getObjectName());
                    alarmObjectMap.put(object.getAlarmId(), objectNameList);
                }
            }

            for (BdpRealTaskAlarm record : page.getRecords()) {
                //状态
                record.setStateFlage(Objects.equals(record.getState(),cn.getech.data.intelligence.common.utils.Constant.StateEnum.ENABLE.getValue()));
                receivedNameList = receivedMap.get(record.getId());
                if (CollectionUtil.isNotEmpty(receivedNameList)) {
                    record.setReceivedNames(String.join(",", receivedNameList));
                }
                objectNameList = alarmObjectMap.get(record.getId());
                if (CollectionUtil.isNotEmpty(objectNameList)) {
                    record.setObjectNames(String.join(",", objectNameList));
                }
                //触发方式
                Constant.AlarmStartModeEnum[] values = Constant.AlarmStartModeEnum.values();
                String triggerConditionTatil = "";
                for (Constant.AlarmStartModeEnum value : values) {
                    if(Objects.equals(record.getTriggerCondition(),value.getCode())){
                        triggerConditionTatil = value.getName();
                    }
                }
                //启用和禁用状态

                record.setTriggerConditionTatil(triggerConditionTatil);
                //任务类型
                record.setTaskTypeTatil(cn.getech.data.development.constant.Constant.RealTimeTaskType.getNameByCode(record.getTaskType()));
                String alarmMode = record.getAlarmMode();
                if (!StringUtils.isEmpty(alarmMode)) {
                    String[] split = alarmMode.split(",");
                    List<String> modeNames = new ArrayList<>();
                    for (String s : split) {
                        try {
                            String s1 = Constant.AlarmModeEnum.ofByCode(Integer.valueOf(s));
                            if (!StringUtils.isEmpty(s1)) {
                                modeNames.add(s1);
                            }
                        } catch (Exception e) {
                            log.error("转换报警方式异常！error:{}", e.getMessage());
                        }
                    }
                    //报警方式
                    record.setAlarmModeTatil(String.join(",", modeNames));
                }
            }
        }
        return new PageUtils(page);
    }

    @Override
    public void delete(Integer id) {
        bdpRealTaskAlarmMapper.deleteById(id);
        QueryWrapper wrapper = new QueryWrapper();
        wrapper.eq("alarm_id", id);
        // 删除接收人
        iBdpRealTaskAlarmReceivedService.remove(wrapper);
        // 删除告警对象
        iBdpRealTaskAlarmObjectService.remove(wrapper);
        //告警记录标记为已删除
        iBdpRealTaskAlarmSendLogService.deleteLogs(id);
    }

    @Override
    public void updateStatusById(Integer id, Integer state) {
        bdpRealTaskAlarmMapper.updateStatusById(id, state);
        //如果是禁用，则对应的待推送的报警全部删除掉
        if (state == cn.getech.data.intelligence.common.utils.Constant.StateEnum.DISABLE.getValue()) {
            //告警记录标记为已删除
            iBdpRealTaskAlarmSendLogService.deleteLogs(id);
        }
    }

    @Override
    public void save(RealTaskAlarmSaveDto realTaskAlarmSaveDto) {
        //验证对应的数据
        cheackFormData(realTaskAlarmSaveDto);

        BdpRealTaskAlarm bdpRealTaskAlarm = new BdpRealTaskAlarm();
        BeanUtils.copyProperties(realTaskAlarmSaveDto, bdpRealTaskAlarm);
        int userId = ShiroUtils.getUserId().intValue();
        bdpRealTaskAlarm.setCreatePer(userId);
        bdpRealTaskAlarm.setTenantId(ShiroUtils.getTenantId());
        if (CollectionUtil.isNotEmpty(realTaskAlarmSaveDto.getAlarmModes())) {
            bdpRealTaskAlarm.setAlarmMode(Joiner.on(",").join(realTaskAlarmSaveDto.getAlarmModes()));
        }
        //默认启用
        bdpRealTaskAlarm.setState(cn.getech.data.intelligence.common.utils.Constant.StateEnum.ENABLE.getValue());
        bdpRealTaskAlarmMapper.insert(bdpRealTaskAlarm);
        //接收人信息
        List<BdpRealTaskAlarmReceived> receivedList = pingAllReceivedList(realTaskAlarmSaveDto, userId, bdpRealTaskAlarm.getId());
        if (CollectionUtil.isNotEmpty(receivedList)) {
            iBdpRealTaskAlarmReceivedService.saveBatch(receivedList);
        }
        //报警对象
        List<Integer> alarmObjects = realTaskAlarmSaveDto.getAlarmObjects();
        if (CollectionUtil.isNotEmpty(alarmObjects)) {
            //查询到所有的实时任务根据id
            List<RealTimeTaskMenu> olderLists = (List<RealTimeTaskMenu>) realTimeTaskMenuService.listByIds(alarmObjects);
            if(CollectionUtil.isEmpty(olderLists)){
                throw new RRException("报警对象有误，请重试！");
            }
            List<BdpRealTaskAlarmObject> lists = new ArrayList<>();
            for (RealTimeTaskMenu realTimeTaskMenu : olderLists) {
                BdpRealTaskAlarmObject bdpRealTaskAlarmObject = new BdpRealTaskAlarmObject();
                bdpRealTaskAlarmObject.setAlarmId(bdpRealTaskAlarm.getId());
                bdpRealTaskAlarmObject.setObjectType(bdpRealTaskAlarm.getTaskType());
                bdpRealTaskAlarmObject.setObjectId(realTimeTaskMenu.getId());
                bdpRealTaskAlarmObject.setObjectName(realTimeTaskMenu.getName());
                lists.add(bdpRealTaskAlarmObject);
            }
            iBdpRealTaskAlarmObjectService.saveBatch(lists);
        }
    }

    private List<BdpRealTaskAlarmReceived> pingAllReceivedList(RealTaskAlarmSaveDto realTaskAlarmSaveDto, Integer userId, Integer id) {
        List<BdpRealTaskAlarmReceived> receivedList = new ArrayList<>();
        if (Objects.equals(Constant.ReceivedTypeEnum.SELF.getCode(), realTaskAlarmSaveDto.getReceivedType())) {
            //本人
            BdpRealTaskAlarmReceived bdpRealTaskAlarmReceived = new BdpRealTaskAlarmReceived();
            bdpRealTaskAlarmReceived.setAlarmId(id);
            bdpRealTaskAlarmReceived.setReceivedId(userId);
            SysUserEntity sysUserEntity = sysUserService.getById(userId);
            bdpRealTaskAlarmReceived.setReceivedName(null == sysUserEntity ? null : sysUserEntity.getUsername());
            bdpRealTaskAlarmReceived.setCreatePer(userId);
            receivedList.add(bdpRealTaskAlarmReceived);
        } else {
            List<BdpRealTaskAlarmReceived> receivedList1 = realTaskAlarmSaveDto.getReceivedList();
            if (CollectionUtil.isNotEmpty(receivedList1)) {
                for (BdpRealTaskAlarmReceived bdpRealTaskAlarmReceived : receivedList1) {
                    bdpRealTaskAlarmReceived.setAlarmId(id);
                    bdpRealTaskAlarmReceived.setCreatePer(userId);
                    receivedList.add(bdpRealTaskAlarmReceived);
                }
            }
        }
        return receivedList;
    }

    @Override
    public void update(RealTaskAlarmSaveDto realTaskAlarmSaveDto) {


        cheackFormData(realTaskAlarmSaveDto);
        if (null == realTaskAlarmSaveDto.getId()) {
            throw new RRException("告警id必须传递！");
        }
        BdpRealTaskAlarm bdpRealTaskAlarm = bdpRealTaskAlarmMapper.selectById(realTaskAlarmSaveDto.getId());
        if (null == bdpRealTaskAlarm) {
            throw new RRException("报警规则不存在,请重新刷新数据！");
        }
        //只能修改名字、触发方式、免打扰时间、报警方式、任务对象、接收人类型
        bdpRealTaskAlarm.setName(realTaskAlarmSaveDto.getName());
        //判断触发报警的条件是否改变。如果有改变，则对应的报警记录也要删除
        if(!Objects.equals(realTaskAlarmSaveDto.getTriggerCondition(),bdpRealTaskAlarm.getTriggerCondition())){
            iBdpRealTaskAlarmSendLogService.deleteLogsByTriggerCondition(bdpRealTaskAlarm.getId(), bdpRealTaskAlarm.getTriggerCondition());
        }
        bdpRealTaskAlarm.setTriggerCondition(realTaskAlarmSaveDto.getTriggerCondition());
        if (CollectionUtil.isNotEmpty(realTaskAlarmSaveDto.getAlarmModes())) {
            bdpRealTaskAlarm.setAlarmMode(Joiner.on(",").join(realTaskAlarmSaveDto.getAlarmModes()));
        }
        bdpRealTaskAlarm.setNoDisturbingStartTime(realTaskAlarmSaveDto.getNoDisturbingStartTime());
        bdpRealTaskAlarm.setNoDisturbingEntTime(realTaskAlarmSaveDto.getNoDisturbingEntTime());
        bdpRealTaskAlarm.setReceivedType(realTaskAlarmSaveDto.getReceivedType());
        bdpRealTaskAlarmMapper.updateById(bdpRealTaskAlarm);
        //置空时间
        if (null == realTaskAlarmSaveDto.getNoDisturbingStartTime() && null == realTaskAlarmSaveDto.getNoDisturbingEntTime()) {
            bdpRealTaskAlarmMapper.updateNoDisturbingNull(realTaskAlarmSaveDto.getId());
        }
        //清空对应的任务对象、接收人类型 并且保存
        QueryWrapper wrapper = new QueryWrapper();
        wrapper.eq("alarm_id", realTaskAlarmSaveDto.getId());
        iBdpRealTaskAlarmReceivedService.remove(wrapper);

        //接收人信息
        int userId = ShiroUtils.getUserId().intValue();
        List<BdpRealTaskAlarmReceived> receivedList = pingAllReceivedList(realTaskAlarmSaveDto, userId, bdpRealTaskAlarm.getId());
        if (CollectionUtil.isNotEmpty(receivedList)) {
            iBdpRealTaskAlarmReceivedService.saveBatch(receivedList);
        }
        //报警对象
        List<BdpRealTaskAlarmObject> olderObjectLists = iBdpRealTaskAlarmObjectService.list(wrapper);
        iBdpRealTaskAlarmObjectService.remove(wrapper);
        //报警对象
        List<Integer> alarmObjects = realTaskAlarmSaveDto.getAlarmObjects();
        //挑选出来，删除了那些对应的报警对象。对应的待发邮件的历史表也要删除掉
        List<Integer> deleteTaskMenuIds = new ArrayList<>();
        for (BdpRealTaskAlarmObject olderObjectList : olderObjectLists) {
            if(!alarmObjects.contains(olderObjectList.getObjectId())){
                deleteTaskMenuIds.add(olderObjectList.getObjectId());
            }
        }
        if (CollectionUtil.isNotEmpty(deleteTaskMenuIds)) {
            iBdpRealTaskAlarmSendLogService.deleteLogs(bdpRealTaskAlarm.getId(), deleteTaskMenuIds);
        }
        if (CollectionUtil.isNotEmpty(alarmObjects)) {
            //查询到所有的实时任务根据id
            List<RealTimeTaskMenu> olderLists = (List<RealTimeTaskMenu>) realTimeTaskMenuService.listByIds(alarmObjects);
            if(CollectionUtil.isEmpty(olderLists)){
                throw new RRException("报警对象有误，请重试！");
            }
            List<BdpRealTaskAlarmObject> lists = new ArrayList<>();
            for (RealTimeTaskMenu realTimeTaskMenu : olderLists) {
                BdpRealTaskAlarmObject bdpRealTaskAlarmObject = new BdpRealTaskAlarmObject();
                bdpRealTaskAlarmObject.setAlarmId(realTaskAlarmSaveDto.getId());
                bdpRealTaskAlarmObject.setObjectType(realTaskAlarmSaveDto.getTaskType());
                bdpRealTaskAlarmObject.setObjectId(realTimeTaskMenu.getId());
                bdpRealTaskAlarmObject.setObjectName(realTimeTaskMenu.getName());
                lists.add(bdpRealTaskAlarmObject);
            }
            iBdpRealTaskAlarmObjectService.saveBatch(lists);
        }

    }

    private void cheackFormData(RealTaskAlarmSaveDto realTaskAlarmSaveDto) {
        if (null == realTaskAlarmSaveDto) {
            throw new RRException("提交的表单数据错误！");
        }
        if (StringUtils.isEmpty(realTaskAlarmSaveDto.getName())) {
            throw new RRException("报警名称必须填写！");
        }
        if (realTaskAlarmSaveDto.getName().length() > 20) {
            throw new RRException("报警名称必须小于20！");
        }
        if (null == realTaskAlarmSaveDto.getProcId()) {
            throw new RRException("报警规则对应的项目必须填写！");
        }
        if (null == realTaskAlarmSaveDto.getTaskType()) {
            throw new RRException("报警规则对应的报警类型必须填写！");
        }
        if (CollectionUtil.isEmpty(realTaskAlarmSaveDto.getAlarmObjects())) {
            throw new RRException("报警规则对应的报警对象必须填写！");
        }
        if (null == realTaskAlarmSaveDto.getTriggerCondition()) {
            throw new RRException("报警规则对应的触发方式必须填写！");
        }
        if (CollectionUtil.isEmpty(realTaskAlarmSaveDto.getAlarmModes())) {
            throw new RRException("报警规则对应的报警方式必须填写！");
        }
        if (null == realTaskAlarmSaveDto.getReceivedType()) {
            throw new RRException("报警规则对应的接收人类型必须填写！");
        }
        if (!Objects.equals(Constant.ReceivedTypeEnum.SELF.getCode(), realTaskAlarmSaveDto.getReceivedType())) {
            if (CollectionUtil.isEmpty(realTaskAlarmSaveDto.getReceivedList())) {
                throw new RRException("报警规则对应的接收人必须填写！");
            }
        }
        //开始时间和结束时间
        if (null != realTaskAlarmSaveDto.getNoDisturbingEntTime() && null == realTaskAlarmSaveDto.getNoDisturbingStartTime()) {
            throw new RRException("免打扰时间，没有开始时间，有结束时间！");
        }
        if (null == realTaskAlarmSaveDto.getNoDisturbingEntTime() && null != realTaskAlarmSaveDto.getNoDisturbingStartTime()) {
            throw new RRException("免打扰时间，有开始时间，没有结束时间！");
        }
        //并且结束时间一定比开始时间大
        if(null != realTaskAlarmSaveDto.getNoDisturbingEntTime() && null != realTaskAlarmSaveDto.getNoDisturbingStartTime()){
            if(realTaskAlarmSaveDto.getNoDisturbingEntTime().compareTo(realTaskAlarmSaveDto.getNoDisturbingStartTime()) <= 0){
                throw new RRException("免打扰时间，结束时间必须大于开始时间！");
            }
        }
    }


    @Override
    public RealTaskAlarmVO getInfoById(Integer id) {
        Long userId = ShiroUtils.getUserId();
        BdpRealTaskAlarm bdpRealTaskAlarm = bdpRealTaskAlarmMapper.selectById(id);
        if (null == bdpRealTaskAlarm) {
            throw new RRException("报警规则不存在,请重新刷新数据！");
        }
        RealTaskAlarmVO realTaskAlarmVO = new RealTaskAlarmVO();
        BeanUtils.copyProperties(bdpRealTaskAlarm, realTaskAlarmVO);
        String[] split = bdpRealTaskAlarm.getAlarmMode().split(",");
        realTaskAlarmVO.setAlarmModes(Arrays.asList(split));
        //项目名称
        ProcInfo procInfo = procInfoService.getById(bdpRealTaskAlarm.getProcId());
        if(null != procInfo){
            realTaskAlarmVO.setProcName(procInfo.getName());
        }
        //触发方式
        Constant.AlarmStartModeEnum[] values = Constant.AlarmStartModeEnum.values();
        String triggerConditionTatil = "";
        for (Constant.AlarmStartModeEnum value : values) {
            if(Objects.equals(bdpRealTaskAlarm.getTriggerCondition(),value.getCode())){
                triggerConditionTatil = value.getName();
            }
        }
        realTaskAlarmVO.setTriggerConditionTatil(triggerConditionTatil);

        //任务类型
        realTaskAlarmVO.setTaskTypeTatil(cn.getech.data.development.constant.Constant.RealTimeTaskType.getNameByCode(bdpRealTaskAlarm.getTaskType()));

        String alarmMode = bdpRealTaskAlarm.getAlarmMode();
        if (!StringUtils.isEmpty(alarmMode)) {
            String[] split1 = alarmMode.split(",");
            List<String> modeNames = new ArrayList<>();
            for (String s : split1) {
                try {
                    String s1 = Constant.AlarmModeEnum.ofByCode(Integer.valueOf(s));
                    if (!StringUtils.isEmpty(s1)) {
                        modeNames.add(s1);
                    }
                } catch (Exception e) {
                    log.error("转换报警方式异常！error:{}", e.getMessage());
                }
            }
            //报警方式
            realTaskAlarmVO.setAlarmModeTatil(String.join(",", modeNames));
        }

        realTaskAlarmVO.setAllProcInfoLists(allProcInfoLists(userId, false));
        realTaskAlarmVO.setAllTaskTypeLists(allTaskTypeLists());
        realTaskAlarmVO.setAllConditionLists(allConditionLists());
        realTaskAlarmVO.setAllAlarmModeLists(allAlarmModeLists());
        realTaskAlarmVO.setAllReceivedTypeLists(allReceivedTypeLists());

        QueryWrapper wrapper = new QueryWrapper();
        wrapper.eq("alarm_id", id);
        List<BdpRealTaskAlarmObject> objectLists = iBdpRealTaskAlarmObjectService.list(wrapper);
        realTaskAlarmVO.setCheackObjectLists(objectLists);
        List<BdpRealTaskAlarmReceived> receivedList = iBdpRealTaskAlarmReceivedService.list(wrapper);
        realTaskAlarmVO.setCheackReceivedLists(receivedList);

        List<Integer> alarmIds = new ArrayList<>();
        alarmIds.add(id);
        List<BdpRealTaskAlarmReceived> receiveds = iBdpRealTaskAlarmReceivedService.listAndNameByIds(alarmIds);
        List<String> receivedNameList = new ArrayList<>();
        if (CollectionUtil.isNotEmpty(receiveds)) {
            receivedNameList = receiveds.stream().map(BdpRealTaskAlarmReceived :: getReceivedName).collect(Collectors.toList());
        }
        List<BdpRealTaskAlarmObject> alarmObjects = iBdpRealTaskAlarmObjectService.listAndNameByIds(alarmIds);
        List<String> objectNameList = new ArrayList<>();
        if (CollectionUtil.isNotEmpty(alarmObjects)) {
            objectNameList = alarmObjects.stream().map(BdpRealTaskAlarmObject :: getObjectName).collect(Collectors.toList());
        }

        if (CollectionUtil.isNotEmpty(receivedNameList)) {
            realTaskAlarmVO.setReceivedNames(String.join(",", receivedNameList));
        }
        if (CollectionUtil.isNotEmpty(objectNameList)) {
            realTaskAlarmVO.setObjectNames(String.join(",", objectNameList));
        }

        return realTaskAlarmVO;
    }

    @Override
    public RealTaskAlarmVO needInfo() {
        Long userId = ShiroUtils.getUserId();
        RealTaskAlarmVO realTaskAlarmVO = new RealTaskAlarmVO();
        realTaskAlarmVO.setAllProcInfoLists(allProcInfoLists(userId, false));
        realTaskAlarmVO.setAllTaskTypeLists(allTaskTypeLists());
        realTaskAlarmVO.setAllConditionLists(allConditionLists());
        realTaskAlarmVO.setAllAlarmModeLists(allAlarmModeLists());
        realTaskAlarmVO.setAllReceivedTypeLists(allReceivedTypeLists());
        return realTaskAlarmVO;
    }

    @Override
    public void cheackIsAlarmAndPushMsg(Integer taskMenuId, Integer runStatus) {
        log.info("开始检测实时任务的报警情况taskMenuId：{},runStatus:{}",taskMenuId,runStatus);
        //查询到这个任务配置的报警信息
        List<RealTaskAlarmInfoDto> alarms = bdpRealTaskAlarmMapper.listByTaskMenuId(taskMenuId);
        if(CollectionUtil.isEmpty(alarms)){
            return;
        }
        //触发报警的条件了,按照收件的邮箱来聚合(目前只支持发送邮件告警)
        Map<String,List<RealTaskAlarmInfoDto>> alarmMaps = new HashMap<>();
        for (RealTaskAlarmInfoDto alarm : alarms) {
            if (Objects.equals(runStatus, alarm.getTriggerCondition())) {
                String email = alarm.getEmail();
                if(!StringUtils.isEmpty(email)){
                    List<RealTaskAlarmInfoDto> realTaskAlarmInfoDtos;
                    if(alarmMaps.containsKey(email)){
                        realTaskAlarmInfoDtos = alarmMaps.get(email);
                    }else{
                        realTaskAlarmInfoDtos = new ArrayList<>();
                    }
                    realTaskAlarmInfoDtos.add(alarm);
                    alarmMaps.put(email,realTaskAlarmInfoDtos);
                }
            }
        }
        //发送报警邮件
        if(MapUtil.isNotEmpty(alarmMaps)){
            ExecutorService es1 = Executors.newSingleThreadExecutor();
            es1.execute(
                    ()-> {
                        genAlarmInfoMsg(alarmMaps);
                    }
            );
        }
    }

    @Override
    public List<RealTaskAlarmInfoDto> cheackAndPush(List<RealTaskAlarmInfoDto> needPushs, LocalTime now) {
        List<RealTaskAlarmInfoDto> needSendLists = new ArrayList<>();
        List<RealTaskAlarmInfoDto> deplaySendLists = new ArrayList<>();//暂时不处理，后面定时处理
        if(CollectionUtil.isNotEmpty(needPushs)){
            firltNeedData(needSendLists,deplaySendLists,needPushs,now);
            if(CollectionUtil.isNotEmpty(needSendLists)){
                //触发报警的条件了,按照收件的邮箱来聚合(目前只支持发送邮件告警)
                Map<String,List<RealTaskAlarmInfoDto>> alarmMaps = needSendLists.stream().collect(Collectors.groupingBy(RealTaskAlarmInfoDto :: getEmail));
                //发送报警邮件
                if(MapUtil.isNotEmpty(alarmMaps)){
                    ExecutorService es1 = Executors.newSingleThreadExecutor();
                    es1.execute(
                            ()-> {
                                initPushAlarmInfoMsg(alarmMaps);
                            }
                    );
                }
            }
        }
        return needSendLists;
    }

    @Override
    public List<IdAndNameDto> getRecriverList() {
        return baseMapper.getRecriverList(ShiroUtils.getTenantId());
    }

    @Override
    public List<IdAndNameDto> getCreaterList() {
        return baseMapper.getCreaterList(ShiroUtils.getTenantId());
    }

    private void initPushAlarmInfoMsg(Map<String, List<RealTaskAlarmInfoDto>> alarmMaps) {
        if(MapUtil.isNotEmpty(alarmMaps)){
            for(Map.Entry<String,List<RealTaskAlarmInfoDto>> entry : alarmMaps.entrySet()){
                List<RealTaskAlarmInfoDto> alarms = entry.getValue();
                if(CollectionUtil.isEmpty(alarms)){
                    continue;
                }
                String toEmail = entry.getKey();
                try {
                    sendEmails(toEmail,alarms);
                }catch (Exception e){
                    log.error("发送实时任务告警邮件异常！error:{}",e.getMessage());

                }
            }
        }
    }

    private void firltNeedData(List<RealTaskAlarmInfoDto> needSendLists,
                               List<RealTaskAlarmInfoDto> deplaySendLists,
                               List<RealTaskAlarmInfoDto> needPushs, LocalTime now) {
        if(CollectionUtil.isNotEmpty(needPushs)){
            for (RealTaskAlarmInfoDto alarmInfoDto : needPushs) {
                LocalTime noDisturbingStartTime = alarmInfoDto.getNoDisturbingStartTime();
                LocalTime noDisturbingEntTime = alarmInfoDto.getNoDisturbingEntTime();
                if(null != noDisturbingStartTime && null != noDisturbingEntTime){
                    try {
                        if(now.compareTo(noDisturbingStartTime) >= 0 && noDisturbingEntTime.compareTo(now) >= 0){
                            deplaySendLists.add(alarmInfoDto);
                            continue;
                        }
                    }catch (Exception e){
                        log.error("转换实时报警信息免打扰的开始时间和结束时间异常！error:{}",e.getMessage());
                    }
                }
                needSendLists.add(alarmInfoDto);
            }
        }
    }

    /***
     * 生成报警信息并且推送报警
     * @param alarmMaps 报警参数---目前只支持邮件报警
     */
    private void genAlarmInfoMsg(Map<String,List<RealTaskAlarmInfoDto>> alarmMaps) {
        LocalTime now = LocalTime.now();
        LocalDateTime nowTime = LocalDateTime.now();
        if(MapUtil.isNotEmpty(alarmMaps)){
            for(Map.Entry<String,List<RealTaskAlarmInfoDto>> entry : alarmMaps.entrySet()){
                List<RealTaskAlarmInfoDto> alarms = entry.getValue();
                if(CollectionUtil.isEmpty(alarms)){
                    continue;
                }
                String toEmail = entry.getKey();
                try {
                    //挑选出来马上需要发的和有免打扰时间，并且在免打扰时间以外的数据
                    List<RealTaskAlarmInfoDto> needSendLists = new ArrayList<>();
                    List<RealTaskAlarmInfoDto> deplaySendLists = new ArrayList<>();
                    firltData(needSendLists,deplaySendLists,alarms,now,nowTime);
                    //免打扰时间内的报警，记录到报警表中
                    if(CollectionUtil.isNotEmpty(deplaySendLists)){
                        batchAllData(toEmail,deplaySendLists);
                    }
                    //直接发送的报警信息，直接发送
                    if(CollectionUtil.isNotEmpty(needSendLists)){
                        sendEmails(toEmail,needSendLists);
                    }
                }catch (Exception e){

                    log.error("发送实时任务告警邮件异常！error:{}",e.getMessage());
                }
            }
        }
    }

    private void sendEmails(String toEmail, List<RealTaskAlarmInfoDto> needSendLists) throws Exception {
        HashMap<String,String> map = new HashMap<>();
        if(CollectionUtil.isNotEmpty(needSendLists)){
            RealTimeTaskVO byMenuId1 = realTimeTaskMapper.getByMenuId(needSendLists.get(0).getObjectId());
            //错误日志信息
            map.put("logInfo",queryRealLog(byMenuId1.getJobId()));
            map.put("title","【实时任务告警通知】");
            map.put("alarmName","实时任务运行通知"); //问题类别
            for (RealTaskAlarmInfoDto needSendList : needSendLists) {
                map.put("alarmName",needSendList.getName()); //告警名称
                map.put("objectName",needSendList.getObjectName()); //项目名称
                Integer taskType = needSendList.getTaskType();
                if(taskType!=null&&taskType==1){
                    map.put("taskType","FLINK API"); //任务类型
                }
                if(taskType!=null&&taskType==1){
                    map.put("taskType","FLINK SQL"); //任务类型
                }
                if(taskType!=null&&taskType==1){
                    map.put("taskType","FLINK SQL DAG"); //任务类型
                }
                if(taskType!=null&&taskType==1){
                    map.put("taskType","实时采集"); //任务类型
                }
                UserListRes info = sysUserService.info(needSendList.getCreatePer().longValue());
                map.put("createPer",info.getUsername()); //创建人
            }
            StringBuilder stringBuilder = tableToHTML(map);
            sendEMailUtil.createSimpleMail(toEmail,"【实时任务告警通知】",stringBuilder.toString());
        }
    }

    private void batchAllData(String toEmail, List<RealTaskAlarmInfoDto> deplaySendLists) {
        //待推送
        if(CollectionUtil.isNotEmpty(deplaySendLists)){
            List<BdpRealTaskAlarmSendLog> lists = new ArrayList<>();
            for (RealTaskAlarmInfoDto realTaskAlarmInfoDto : deplaySendLists) {
                BdpRealTaskAlarmSendLog bdpRealTaskAlarmSendLog = new BdpRealTaskAlarmSendLog();
                bdpRealTaskAlarmSendLog.setCreateTime(realTaskAlarmInfoDto.getAlarmTime());
                bdpRealTaskAlarmSendLog.setAlarmId(realTaskAlarmInfoDto.getId());
                bdpRealTaskAlarmSendLog.setRealTaskMenuId(realTaskAlarmInfoDto.getObjectId());
                bdpRealTaskAlarmSendLog.setIsDelete(0);
                bdpRealTaskAlarmSendLog.setIsSend(0);
                lists.add(bdpRealTaskAlarmSendLog);
            }
            iBdpRealTaskAlarmSendLogService.saveBatch(lists);
        }
    }

    private void firltData(List<RealTaskAlarmInfoDto> needSendLists,
                           List<RealTaskAlarmInfoDto> deplaySendLists,
                           List<RealTaskAlarmInfoDto> alarms,
                           LocalTime now, LocalDateTime nowTime) {
        if(CollectionUtil.isNotEmpty(alarms)){
            for (RealTaskAlarmInfoDto alarmInfoDto : alarms) {
                LocalTime noDisturbingStartTime = alarmInfoDto.getNoDisturbingStartTime();
                LocalTime noDisturbingEntTime = alarmInfoDto.getNoDisturbingEntTime();
                alarmInfoDto.setAlarmTime(nowTime);//报警时间
                if(null != noDisturbingStartTime && null != noDisturbingEntTime){
                    try {
                        if(now.compareTo(noDisturbingStartTime) >= 0 && noDisturbingEntTime.compareTo(now) >= 0){
                            deplaySendLists.add(alarmInfoDto);
                            continue;
                        }
                    }catch (Exception e){
                        log.error("转换实时报警信息免打扰的开始时间和结束时间异常！error:{}",e.getMessage());
                    }
                }
                needSendLists.add(alarmInfoDto);
            }
        }
    }

    private List<ProcInfo> allProcInfoLists(Long userId, Boolean isAdmin) {
        if (isAdmin) {
            QueryWrapper wrapper = new QueryWrapper();
            wrapper.orderByDesc("id");
            return procInfoService.list(wrapper);
        }
        return procInfoService.listUserProcs(userId);
    }

    private List<IdAndNameVO> allTaskTypeLists() {
        List<IdAndNameVO> allReceivedTypeLists = new ArrayList<>();
        for (cn.getech.data.development.constant.Constant.RealTimeTaskType value : cn.getech.data.development.constant.Constant.RealTimeTaskType.values()) {
            IdAndNameVO idAndNameDto = new IdAndNameVO();
            idAndNameDto.setId(value.getCode());
            idAndNameDto.setName(value.getName());
            allReceivedTypeLists.add(idAndNameDto);
        }
        return allReceivedTypeLists;
    }

    private List<IdAndNameVO> allConditionLists() {
        List<IdAndNameVO> allReceivedTypeLists = new ArrayList<>();
        for (RealTaskAlarmModeEnum value : RealTaskAlarmModeEnum.values()) {
            IdAndNameVO idAndNameDto = new IdAndNameVO();
            idAndNameDto.setId(value.getCode());
            idAndNameDto.setName(value.getName());
            allReceivedTypeLists.add(idAndNameDto);
        }
        return allReceivedTypeLists;
    }

    private List<IdAndNameVO> allAlarmModeLists() {
        List<IdAndNameVO> allReceivedTypeLists = new ArrayList<>();
        for (Constant.AlarmModeEnum value : Constant.AlarmModeEnum.values()) {
            IdAndNameVO idAndNameDto = new IdAndNameVO();
            idAndNameDto.setId(value.getCode());
            idAndNameDto.setName(value.getName());
            allReceivedTypeLists.add(idAndNameDto);
        }
        return allReceivedTypeLists;
    }

    private List<IdAndNameVO> allReceivedTypeLists() {
        List<IdAndNameVO> allReceivedTypeLists = new ArrayList<>();
        for (Constant.ReceivedTypeEnum value : Constant.ReceivedTypeEnum.values()) {
            IdAndNameVO idAndNameDto = new IdAndNameVO();
            idAndNameDto.setId(value.getCode());
            idAndNameDto.setName(value.getName());
            allReceivedTypeLists.add(idAndNameDto);
        }
        return allReceivedTypeLists;
    }

    /**
     * 获取实时任务日志
     * @param appId
     * @return
     */
   public String  queryRealLog(String appId){
       if (org.apache.commons.lang.StringUtils.isNotEmpty(appId) && appId.contains("/usr/realtime/")){
           ExecuteShellUtil resShell = ExecuteShellUtil.getInstance();
           try {
               resShell.init(developmentConfig.getFLinkHost(), developmentConfig.getFLinkPort(),
                       developmentConfig.getFLinkUserName(), developmentConfig.getFLinkPwd());
               String res = resShell.getFile(appId);
               if (res.length() > 200000) {
                   res = res.substring(0, 200000);
               }
               res = new String(res.getBytes(StandardCharsets.UTF_8));
               res = new String(res.getBytes(StandardCharsets.UTF_8),StandardCharsets.UTF_8);
               log.info("运行日志:\n{}",res);
               return res;
           } catch (JSchException e) {
               return null;
           }
       }
       R r = getJobId(appId);
       if (!r.isOk()) {
           return JSONObject.toJSONString(geiHdfsLog(appId));
       }
       try {
           String jobId = r.get("data").toString();
           ///jobs/cecb847460bffa34e36c1a2419eaf174/exceptions
           String url = developmentConfig.getFLinkImpl() + appId + "/jobs/" + jobId + "/exceptions";
           String data = HttpUtil.get(url, 5000);
           JSONObject check = JSONObject.parseObject(data);
           if (check.keySet().contains("root-exception") &&
                   org.apache.commons.lang.StringUtils.isNotEmpty(check.getString("root-exception"))
           ) {
               return check.toString();
           } else {
               return JSONObject.toJSONString(geiHdfsLog(appId));
           }
       } catch (Exception e) {
           log.error("Flink错误:{}", e.getMessage());
           return JSONObject.toJSONString(geiHdfsLog(appId));
       }
    }
    private JSONObject geiHdfsLog(String appId){
        JSONObject dataRes = new JSONObject();
        for (String defaultFS : bdpJobConfig.getNamenodestr().split(",")) {
            HadoopLogUtils.setDefaultFS(defaultFS);
            try {
                String s = HadoopLogUtils.printLog(appId, "hdfs");
                if (org.apache.commons.lang.StringUtils.isNotEmpty(s)) {
                    dataRes.put("root-exception", s);
                    break;
                }
            } catch (Exception e) {
                log.warn("获取失败:{} appId:{}", defaultFS, appId, e);
            }
        }
        return dataRes;
    }
    private R getJobId(String appId) {
        if (org.apache.commons.lang.StringUtils.isEmpty(appId)) {
            return R.error("应用id不能为空！");
        }
        String jobId = null;
        try {
            //application_1600679071671_15393/jobs/overview
            String url = developmentConfig.getFLinkImpl() + appId + "/jobs/overview";
            String data = HttpUtil.get(url, 5000);
            JSONObject job = JSONObject.parseObject(data);
            JSONArray jobs = job.getJSONArray("jobs");
            if (jobs != null && jobs.size() > 0) {
                jobId = jobs.getJSONObject(0).getString("jid");
            }
        } catch (Exception e) {
            return R.error("Hadoop不存在【" + appId + "】任务，请稍后尝试");
        }
        if (jobId == null) {
            return R.error("任务不存在！");
        }
        return R.okWithData(jobId);
    }

    /**
     * 实时任务 邮件发送
     * @param stringHashMap
     * @return
     */
    public static StringBuilder tableToHTML( HashMap<String,String> stringHashMap){
        DateTimeFormatter dtf = DateTimeFormatter.ofPattern("yyyy年-MM月-dd日 HH时:mm分:ss秒");
        String formatTimeStr = dtf.format(ZonedDateTime.now());
        StringBuilder content = new StringBuilder("<html><head></head>" +
                "<body><h2>"+stringHashMap.get("title")+"</h2>");
        content.append("<div>");
        content.append("邮件标题  实时任务告警").append("<br>");
        content.append("</div>");
        content.append("<div>");
        content.append("邮件内容 实时任务告警 "+formatTimeStr+" 检测到"+stringHashMap.get("alarmName")+"异常").append("<br>");
        content.append("</div>");

        content.append("<table border=\"5\" style=\"border:solid 1px #E8F2F9;font-size=14px;;font-size:18px;\">");
        content.append("<tr style=\"background-color: #428BCA; color:#ffffff\">");
        content.append("<td>" + "问题类别" + "</td>");
        content.append("<td>" + stringHashMap.get("alarmName") + "</td>"); //表别名
        content.append("</tr>");
        content.append("<tr style=\"background-color: #428BCA; color:#ffffff\">");
        content.append("<td>" + "任务名称" + "</td>");
        content.append("<td>" + stringHashMap.get("alarmName") + "</td>"); //表别名
        content.append("</tr>");
        content.append("<tr style=\"background-color: #428BCA; color:#ffffff\">");
        content.append("<td>" + "任务类型" + "</td>");
        content.append("<td>" + stringHashMap.get("taskType") + "</td>"); //表别名
        content.append("</tr>");
        content.append("<tr style=\"background-color: #428BCA; color:#ffffff\">");
        content.append("<td>" + "任务创建人" + "</td>");
        content.append("<td>" + stringHashMap.get("createPer") + "</td>");
        content.append("</tr>");
        content.append("<tr style=\"background-color: #428BCA; color:#ffffff\">");
        content.append("<td>" + "所属项目" + "</td>");
        content.append("<td>" + stringHashMap.get("objectName") + "</td>"); //表别名
        content.append("</tr>");
        content.append("<tr style=\"background-color: #428BCA; color:#ffffff\">");
        content.append("<td>" + "项目创建人" + "</td>");
        content.append("<td>" + stringHashMap.get("createPer") + "</td>"); //表别名
        content.append("</tr>");
        content.append("<tr style=\"background-color: #428BCA; color:#ffffff\">");
        content.append("<td>" + "报错日志" + "</td>");
        content.append("<td>" + stringHashMap.get("logInfo") + "</td>"); //表别名
        content.append("</tr>");
        content.append("</table>");
        return content;
    }
}
